package com.abel.ignite.impl;

import com.abel.ignite.DDLFactory;
import com.abel.ignite.DMLFactory;
import com.abel.ignite.model.RepositoryModel;
import com.abel.ignite.parser.DDLParser;
import lombok.extern.slf4j.Slf4j;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.cache.CacheMode;
import org.apache.ignite.cache.query.FieldsQueryCursor;
import org.apache.ignite.cache.query.SqlFieldsQuery;
import org.apache.ignite.configuration.CacheConfiguration;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
import javax.annotation.Resource;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.Collection;
import java.util.Collections;
import java.util.List;

/**
 * @author xiachao
 * @date 2019-04-28
 */
@Slf4j
@Component
public class RepositoryManager implements DDLFactory, DMLFactory {
    @Resource
    private Ignite ignite;

    private static final String PRE_FFIX = "SQL_PUBLIC_";

    @Override
    public boolean createRepository(RepositoryModel model) {
        if(model == null || StringUtils.isEmpty(model.getRepository())){
            return false;
        }
        String table = model.getRepository();
        if(this.isExsit(table)){
            log.error("Table {} already exsit!", table);
            return false;
        }
        String sql = DDLParser.parseCreate(model);
        if(StringUtils.isEmpty(sql)){
            return false;
        }
        IgniteCache igniteCache = ignite.createCache(model.getRepository());
        igniteCache.query(new SqlFieldsQuery(sql)).getAll();
        return true;
    }

    @Override
    public boolean createRepositoryByJDBC(RepositoryModel model) {
        if(model == null || StringUtils.isEmpty(model.getRepository())){
            return false;
        }
        String sql = DDLParser.parseCreate(model);
        if(StringUtils.isEmpty(sql)){
            return false;
        }
        Connection conn = null;
        try{
            // Register JDBC driver.
            Class.forName("org.apache.ignite.IgniteJdbcThinDriver");

            // Open JDBC connection.
            conn = DriverManager.getConnection("jdbc:ignite:thin://127.0.0.1/");

            // Create database tables.
            try (Statement stmt = conn.createStatement()) {

                // Create table based on REPLICATED template.
                stmt.executeUpdate(sql);
            }
        }catch (Exception e){
            return false;
        }finally {
            if(conn != null){
                try {
                    conn.close();
                } catch (SQLException e) {
                    return false;
                }
            }
        }
        return true;
    }

    public Collection<String> getCacheNames(){
        return this.ignite.cacheNames();
    }

    @Override
    public boolean isExsit(String cacheName) {
        return this.ignite.cacheNames().contains(this.wrap(cacheName));
    }

    @Override
    public boolean dropRepository(String cacheName) {
        if(StringUtils.isEmpty(cacheName)){
            log.error("Repository Name can not be empty!");
            return false;
        }
        StringBuilder sql = new StringBuilder();
        sql.append("DROP TABLE ");
        sql.append(cacheName);
        IgniteCache igniteCache = ignite.createCache(cacheName);
        igniteCache.query(new SqlFieldsQuery(sql.toString())).getAll();
        return false;
    }

    @Override
    public List<List<?>> query(String cacheName, String sql) {
        if(StringUtils.isEmpty(sql)){
            return Collections.emptyList();
        }
        // Querying data from the cluster using a distributed JOIN.
        SqlFieldsQuery query = new SqlFieldsQuery(sql);
        IgniteCache igniteCache = this.ignite.getOrCreateCache(this.wrap(cacheName));
        FieldsQueryCursor<List<?>> cursor = igniteCache.query(query);
        return cursor.getAll();
    }

    @Override
    public boolean insert(String cacheName, String sql) {
        return this.update(cacheName, sql);
    }

    @Override
    public boolean update(String cacheName, String sql) {
        if(StringUtils.isEmpty(sql)){
            return false;
        }
        IgniteCache igniteCache = this.ignite.cache(this.wrap(cacheName));
        igniteCache.query(new SqlFieldsQuery(sql)).getAll();
        return true;
    }

    @Override
    public boolean delete(String cacheName, String sql) {
        return this.update(cacheName, sql);
    }

    private String wrap(String cacheName){
        return PRE_FFIX + cacheName.toUpperCase();
    }
}
